package polimi.disttest.manager;

import java.io.BufferedReader;

import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

import polimi.disttest.LocalManager;
import polimi.disttest.ResultsAnalyzer;
import polimi.reds.optimizer.OptimizerStrategy;

public class RemoteManager {
	private static Logger log = Logger.getLogger(RemoteManager.class);
	private String localjar="/home/ricky/redsoptimizer.jar";
	private String remotejar="/home/ricky/Documents/redsoptimizer.jar";
	private MachineSSHAddress machines[] = new MachineSSHAddress[] {
		new MachineSSHAddress("ricky","127.0.0.1", 22, 20,"nara3366")	
	};
	private boolean updateJars=true;
	private String cmdCommand="java -jar " + remotejar
		+ " -" + LocalManager.OPTION_ALGORITHM + " NullOptimizer"
		+ " -" + LocalManager.OPTION_ALGORITHM_PARAMETERS + " threshold=20,maxneighbors=5,updateevents=100"
		+ " -" + LocalManager.OPTION_CLASSES_PER_SUBSCRIPTION + " 3"
		+ " -" + LocalManager.OPTION_DELAY_MESSAGES_MEAN + " 400"
		+ " -" + LocalManager.OPTION_DELAY_MESSAGES_STDEV + " 100"
		+ " -" + LocalManager.OPTION_DURATION + " 300"
		+ " -" + LocalManager.OPTION_MESSAGE_CLASSES + " 5"
		+ " -" + LocalManager.OPTION_MESSAGES_CLASSES_DISTRIBUTION + " zipf"
		+ " -" + LocalManager.OPTION_NUM_COMPONENT_SUBSCRIPTIONS + " 2"
		+ " -" + LocalManager.OPTION_NUM_COMPONENTS + " 1"
		+ " -" + LocalManager.OPTION_SEED + " 1"
		+ " -" + LocalManager.OPTION_STARTUP_NEIGHBORHOOD_TIMESTAMP + " " + ((int) (System.currentTimeMillis()/1000) + 9)
		+ " -" + LocalManager.OPTION_STARTUP_TRAFFIC_TIMESTAMP + " " + ((int) (System.currentTimeMillis()/1000) + 17)
		+ " -" + LocalManager.OPTION_SIMULID + " " +  ((int) (System.currentTimeMillis()/1000))
		+ " -" + LocalManager.OPTION_SUBSCRIPTIONS_CLASSES_DISTRIBUTION + " zipf";
	private Random rand=new Random(123);
	private List<Edge> links = null;
	private List<Node> nodes = null;
	private ResultsCollector resultsCollector = new ResultsCollector();
	private Map<String,Object> collectedData = new HashMap<String,Object>();
	private String aggregatedStatistics[] = {
			OptimizerStrategy.STAT_ATTEMPTED_RECONFIGURATIONS_COUNT,
			OptimizerStrategy.STAT_SUCCESSFUL_RECONFIGURATIONS_COUNT,
			OptimizerStrategy.STAT_RECONFIGURATION_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_RECONFIGURATION_TRAFFIC_SIZE,
			OptimizerStrategy.STAT_TOTAL_OVERLAY_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_TOTAL_OVERLAY_TRAFFIC_SIZE,
			OptimizerStrategy.STAT_SUB_OVERLAY_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_SUB_OVERLAY_TRAFFIC_SIZE,
			OptimizerStrategy.STAT_PUB_OVERLAY_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_PUB_OVERLAY_TRAFFIC_SIZE,
			OptimizerStrategy.STAT_TOTAL_INCOMING_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_TOTAL_INCOMING_TRAFFIC_SIZE,
			OptimizerStrategy.STAT_SUB_INCOMING_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_SUB_INCOMING_TRAFFIC_SIZE,
			OptimizerStrategy.STAT_PUB_INCOMING_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_PUB_OUTGOING_TRAFFIC_EVENTS,
			OptimizerStrategy.STAT_PUB_OUTGOING_TRAFFIC_SIZE
			};
	
	public static void main(String[] args) {
		PropertyConfigurator.configure("log4j.properties");
		RemoteManager remoteManager = new RemoteManager();
		remoteManager.performSimulation();
	}
	
	public void performSimulation() {
		log.info("*** STARTING SIMULATION ***");
		System.out.println("*** STARTING SIMULATION ***");
		
		// updating jars
		if (updateJars) {
			updateJars();
		}
		
		// creating initial topology
		nodes = new ArrayList<Node>();
		for (MachineSSHAddress machine:machines) {
			for (int i=0; i<machine.getNumBrokers(); i++) {
				nodes.add(new Node(machine,i));
			}
		}
		links = (new ScaleFreeTopology(nodes,rand)).createTopology();
		
		// run simulation on remote nodes
		startRemoteNodes();
		log.info("*** SIMULATION COMPLETED ***");
		System.out.println("*** SIMULATION COMPLETED ***");
		System.exit(0);
	}
	
	private Map<String,Object> getRemoteResults() {
		//List<ProcessWrapper> processes=new ArrayList<ProcessWrapper>();
		for (MachineSSHAddress machine:machines) {
			//ProcessWrapper process=null;
			SSHConnection connection=null;
			//try {
				connection = new SSHConnection(machine.getUsername(), machine.getHost(), machine.getPort(), machine.getPassword());
				connection.createFTP();
				connection.getFile();
				//process = new ProcessWrapper(machine.getUsername() + "@" + machine.getHost() + "/getres", "scp -P " + machine.getPort() + " " + machine.getUsername() + "@" + machine.getHost() + ":*.txt .");
			//	processes.add(process);
//			} catch (IOException e) {
//				System.err.println("Error while starting copying files from remote node (" + e.getMessage() + ")");
//				System.exit(-1);
//			}
		}
		
//		boolean waiting;
//		do {
//			waiting=false;
//			for (ProcessWrapper process:processes) {
//				try {
//					process.printResults();
//					if (process.exitValue()!=0) {
//						System.err.println("Error while copying results files from remote nodes.");
//					}
//						//BufferedReader reader = new BufferedReader(new InputStreamReader(p.getInputStream()));
//						//try {
//						//	while (true) {
//						//		System.err.println(p.hashCode() + " " + reader.readLine());
//						//	}
//						//} catch (IOException e) { }
//						//System.exit(-1);
//						//p.getOutputStream();
//					//}
//				} catch (IllegalThreadStateException e) {
//					waiting=true;
//				}
//				
//				// gather results
//			}
//			try {
//				Thread.sleep(100);
//			} catch (InterruptedException e) {}
//		} while (waiting);
		
		List<Map<String,Object>> data = new ArrayList<Map<String,Object>>();
		for (Node node:nodes) {
			data.add(readData(node.getMachineSSHAddress().getHost() + "." + (10000 + node.getID()) + ".txt"));
		}
		
		ArrayList<String> as=new ArrayList<String>();
		for (String str:aggregatedStatistics) {
			as.add(str);
		}
		
		return sumData(data, as);
	}
	/**
	 * Metodo che avvia la simulazione distribuito
	 */
	public void startRemoteNodes() {
		log.info("Running simulation on remote nodes...");
		System.out.println("Running simulation on remote nodes...");
		//Map<ProcessWrapper,Node> mapProcessToNode = new HashMap<ProcessWrapper,Node>();
		Map<SSHConnection,Node> mapProcessToNode = new HashMap<SSHConnection,Node>();
		for (Node node:nodes) {
			//ProcessWrapper process;
			SSHConnection connection=null;
//			try {
				String initialNeighbors = initialNeighborsURLs(node);
				/*process = new ProcessWrapper(node.getMachineSSHAddress().getUsername() + "@" + node.getMachineSSHAddress().getHost() + "/run" + node.getID().toString(), "ssh " + node.getMachineSSHAddress().sshAddress() + " " + cmdCommand + " -" + LocalManager.OPTION_PORT + " " + (10000 + node.getID()) + " -" + 
						LocalManager.OPTION_IP + " " + node.getMachineSSHAddress().getHost() +
						((initialNeighbors.length()>0) ? (" -" + LocalManager.OPTION_INITIAL_NEIGHBORS + " " + initialNeighbors) : "") +
						" >> node" + node.getID() + ".log 2>&1");*/
				
				connection=new SSHConnection(node.getMachineSSHAddress().getUsername(), node.getMachineSSHAddress().getHost(), node.getMachineSSHAddress().getPort(), node.getMachineSSHAddress().getPassword());
				cmdCommand+=" -" + LocalManager.OPTION_PORT + " " + (10000 + node.getID()) + " -" + 
				LocalManager.OPTION_IP + " " + node.getMachineSSHAddress().getHost();
				if(initialNeighbors.length()>0){
					cmdCommand+=" -" + LocalManager.OPTION_INITIAL_NEIGHBORS + " " + initialNeighbors;
				}else{
					cmdCommand+="";
				}
				cmdCommand+=" >> node" + node.getID() + ".log 2>&1";
				connection.execute(cmdCommand);
				mapProcessToNode.put(connection, node);
//			} catch (IOException e) {
//				System.err.println("Error while starting simulation on remote node (" + e.getMessage() + ")");
//				System.exit(-1);
//			}
		}
		
		try {
			Thread.sleep(10000);
		} catch (InterruptedException e) {}
		
		resultsCollector.start();
		
		boolean waiting;
		int i=0;
/************************************************************************/
/*		do {
			i++;
			waiting=false;
			for (ProcessWrapper process:new ArrayList<ProcessWrapper>(mapProcessToNode.keySet())) {
				try {
					process.printResults();
					if (process.exitValue()!=0) {
						System.err.println("Error during the execution of the remote simulation on node: " + mapProcessToNode.get(process).toString());
						System.exit(-1);
					}
					mapProcessToNode.remove(process);
				} catch (IllegalThreadStateException e) {
					waiting=true;
				}
			}
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {}
			
			collectedData.put("time", i);
			collectedData.remove(LocalManager.OPTION_PORT);
			collectedData.remove(LocalManager.OPTION_DURATION);
			collectedData.remove(LocalManager.OPTION_SEED);
			collectedData.remove(LocalManager.OPTION_MESSAGE_CLASSES);
			collectedData.remove(LocalManager.OPTION_NUM_COMPONENTS);
			collectedData.remove(LocalManager.OPTION_CLASSES_PER_SUBSCRIPTION);
			collectedData.remove(LocalManager.OPTION_NUM_COMPONENT_SUBSCRIPTIONS);
			collectedData.remove(LocalManager.OPTION_STARTUP_TRAFFIC_TIMESTAMP);
			collectedData.remove(LocalManager.OPTION_STARTUP_NEIGHBORHOOD_TIMESTAMP);
			collectedData.remove(LocalManager.OPTION_DELAY_MESSAGES_MEAN);
			collectedData.remove(LocalManager.OPTION_DELAY_MESSAGES_STDEV);
			
			ResultsAnalyzer results = new ResultsAnalyzer(collectedData);
			results.writeResultsToConsole(null);
			
		} while (waiting);
*/	
		log.info("SIMULATION COMPLETED!");
		System.out.println("SIMULATION COMPLETED!");
		
	}
	
	public void updateJars() {
		log.info("Updating jars on remote machines...");
		System.out.println("Updating jars on remote machines...");
		
		//List<ProcessWrapper> processes=new ArrayList<ProcessWrapper>();
		
		for (MachineSSHAddress machine:machines) {
			//ProcessWrapper p;
			SSHConnection connection=null;
			//try {
				connection=new SSHConnection(machine.getUsername(), machine.getHost(), machine.getPort(), machine.getPassword());
				connection.createFTP();
				connection.putFile(localjar,remotejar);
//				p = new ProcessWrapper(machine.getUsername() + "@" + machine.getHost() + "/jarupd", "scp " + "-P " + machine.getPort() + " " + localjar + " " + machine.getUsername() + "@" + machine.getHost() + ":" + remotejar);
//				processes.add(p);
//			} catch (IOException e) {
//				System.err.println("Error while executing scp command (" + e.getMessage() + ")");
//				System.exit(-1);
//			}
			
		}
		
//		boolean waiting;
//		do {
//			waiting=false;
//			for (ProcessWrapper process:processes) {
//				try {
//					process.printResults();
//					if (process.exitValue()!=0) {
//						System.err.println("Error while updating jar! Check addresses and paths!");
//						System.exit(-1);
//					}
//				} catch (IllegalThreadStateException e) {
//					waiting=true;
//				}
//			}
//			try {
//				Thread.sleep(1000);
//			} catch (InterruptedException e) {}
//		} while (waiting);
//		
//		try {
//			Thread.sleep(2000);
//		} catch (InterruptedException e) {}
//		System.out.println("Updating jars: SUCCESSFUL!");
	}
	
	private String initialNeighborsURLs(Node node) {
		List<Node> neighbors = new ArrayList<Node>();
		
		for (Edge link:links) {
			if (link.getNode1().equals(node)) {
				neighbors.add(link.getNode2());
			} //else if (l.getNode2().equals(node)) {
			 //	neighbors.add(l.getNode1());
			//}
		}
		
		String s="";
		for (Node n:neighbors) {
			s += ",reds-tcp:" + n.getMachineSSHAddress().getHost() + ":" + (10000 + n.getID());
		}
		if (s.length() > 0) {
			s=s.substring(1);
		}
		
		return s;
	}
	/**
	 * Metodo utilizzato per aggregare tutti i risultati ricavati
	 * dagli hosts
	 * @param data
	 * @param what
	 * @return
	 */
	private Map<String,Object> sumData(List<Map<String,Object>> data, List<String> what) {
		Map<String,Object> aggregatedData = new HashMap<String,Object>();
		
		for (Map<String,Object> d:data) {
			for (String k:d.keySet()) {
				if (!what.contains(k) || !(d.get(k) instanceof Integer || d.get(k) instanceof Double)) {
					aggregatedData.put(k, d.get(k));
				} else if (d.get(k) instanceof Integer) {
					if (aggregatedData.get(k) == null) {
						aggregatedData.put(k, d.get(k));
					} else {
						aggregatedData.put(k, (Integer) aggregatedData.get(k) + (Integer) d.get(k)); 
					}
				} else if (d.get(k) instanceof Double) {
					if (aggregatedData.get(k) == null) {
						aggregatedData.put(k, d.get(k));
					} else {
						aggregatedData.put(k, (Double) aggregatedData.get(k) + (Double) d.get(k)); 
					}
				}
			}
		}
		
		return aggregatedData;
	}
	
	private Map<String,Object> readData(String fileName) {
		Map<String,Object> map = new HashMap<String,Object>();
		FileReader fileReader=null;
		try {
			fileReader = new FileReader(fileName);
		} catch (FileNotFoundException e) {
			log.error("Unable to open file: " + fileName);
			System.err.println("Unable to open file: " + fileName);
			//System.exit(-1);
			return map;
		}
	    BufferedReader reader = new BufferedReader(fileReader);

	    try {
	      
	    	String line;
	    	do {
	    		line = reader.readLine();
	    		if (line != null && line.length()>0 && line.contains("=")) {
	    			String[] splitLine = line.split("=");
	    			if ((splitLine.length<2) || splitLine[1].length()==0 || splitLine[1].substring(1,1).equals("\""))  {
	    				continue;
	    			}
	    			
	    			try {
	    				map.put(splitLine[0], Double.parseDouble(splitLine[1]));
	    			} catch (NumberFormatException e) {
	    				continue;
	    			}
	    		}
	    		
	    	} while (line != null);
	    	
	      reader.close();
	      fileReader.close();

	    } catch (IOException e) {
	      e.printStackTrace();
	    }
	    
	    return map;
	}
	
	private class ResultsCollector extends Thread {
		
		public ResultsCollector() {
			this.setDaemon(true);
		}
		
		public void run() {
			while (true) {
				try {
					Thread.sleep(5000);
				} catch (InterruptedException e) {
				}
				
				collectedData = getRemoteResults();
			}
		}
	}
	
}